﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Collections;
using System.Threading;
using System.Diagnostics;

namespace MSL.Utility.Threading
{
    /// <summary>
    /// 异步操作类
    /// </summary>
    /// <returns></returns>
    public static class AsyncHelper
    {
        /// <summary> 
        /// 执行多线程操作任务 
        /// </summary> 
        /// <param name="list">多线程操作的数据集合</param> 
        /// <param name="threadNum">分多少个线程来做</param> 
        /// <param name="processMethod">处理数据集合中单个数据使用的处理方法</param> 
        public static void DoAsync(IList list, int threadNum, WaitCallback processMethod)
        {
            DoAsync(list, threadNum, processMethod, true);
        }

        /// <summary> 
        /// 执行多线程操作任务 
        /// </summary> 
        /// <param name="list">多线程操作的数据集合</param> 
        /// <param name="threadNum">分多少个线程来做</param> 
        /// <param name="isNeedWaitAll">是否需要等待所有线程执行完毕才返回，为true时会等待所有线程执行完毕，否则则是在有一个线程执行完毕就返回</param> 
        /// <param name="processMethod">处理数据集合中单个数据使用的处理方法</param> 
        /// <param name="processResult">处理结果</param> 
        public static void DoAsync(IList list, int threadNum, Func<object, object> processMethod, bool isNeedWaitAll, out Hashtable processResult)
        {
            DoAsyncPrivate(list, threadNum, null, processMethod, isNeedWaitAll, true, out processResult);
        }

        /// <summary> 
        /// 执行多线程操作任务 
        /// </summary> 
        /// <param name="list">多线程操作的数据集合</param> 
        /// <param name="threadNum">分多少个线程来做</param> 
        /// <param name="processMethod">处理数据集合中单个数据使用的处理方法</param> 
        /// <param name="isNeedWaitAll">是否需要等待所有线程执行完毕才返回，为true时会等待所有线程执行完毕，否则则是在有一个线程执行完毕就返回</param> 
        public static void DoAsync(IList list, int threadNum, Func<object, object> processMethod, out Hashtable processResult)
        {
            DoAsyncPrivate(list, threadNum, null, processMethod, true, true, out processResult);
        }

        /// <summary> 
        /// 执行多线程操作任务 
        /// </summary> 
        /// <param name="list">多线程操作的数据集合</param> 
        /// <param name="threadNum">分多少个线程来做</param> 
        /// <param name="processMethod">处理数据集合中单个数据使用的处理方法</param> 
        /// <param name="isNeedWaitAll">是否需要等待所有线程执行完毕才返回，为true时会等待所有线程执行完毕，否则则是在有一个线程执行完毕就返回</param> 
        public static void DoAsync(IList list, int threadNum, WaitCallback processMethod, bool isNeedWaitAll)
        {
            Hashtable hash;
            DoAsyncPrivate(list, threadNum, processMethod, null, isNeedWaitAll, false, out hash);
        }

        #region 私有

        private static void DoAsyncPrivate(IList list, int threadNum, WaitCallback processMethod, Func<object, object> getObjMethod, bool isNeedWaitAll, bool hasReturnValue, out Hashtable processResult)
        {
            if (list == null) throw new ArgumentNullException("list");

            if (threadNum >= 64 || threadNum < 2)
            {
                throw new ArgumentOutOfRangeException("threadNum", "threadNum 参数必须在2和64之间");
            }

            if (threadNum > list.Count)
                threadNum = list.Count;

            IList[] colls = new ArrayList[threadNum];

            DataWithStateList dataWithStates = new DataWithStateList();
            AutoResetEvent[] evts = new AutoResetEvent[threadNum];

            for (int i = 0; i < threadNum; i++)
            {
                colls[i] = new ArrayList();
                evts[i] = new AutoResetEvent(false);
            }

            for (int i = 0; i < list.Count; i++)
            {
                object obj = list[i];
                int threadIndex = i % threadNum;
                colls[threadIndex].Add(obj);
                dataWithStates.Add(new DataWithState(obj, ProcessState.WaitForProcess));
            }

            AsyncContext context = AsyncContext.GetContext(threadNum, dataWithStates, isNeedWaitAll, hasReturnValue, processMethod, getObjMethod);

            for (int i = 0; i < threadNum; i++)
            {
                ThreadPool.QueueUserWorkItem(DoPrivate, new object[] {  
                    colls[i],context,evts[i] 
                });
            }

            if (isNeedWaitAll)
            {
                WaitHandle.WaitAll(evts);
            }
            else
            {
                WaitHandle.WaitAny(evts);
                context.SetBreakSignal();
            }
            processResult = context.ProcessResult;
        }

        /// <summary>
        /// 
        /// </summary>
        private class AsyncContext
        {
            static public AsyncContext GetContext(
                int threadNum,
                DataWithStateList dataWithStates,
                bool isNeedWaitAll,
                bool hasReturnValue,
                WaitCallback processMethod,
                Func<object, object> hasReturnValueMethod
                )
            {
                AsyncContext context = new AsyncContext();
                context.ThreadCount = threadNum;
                context.DataWithStates = dataWithStates;
                context.NeedWaitAll = isNeedWaitAll;
                if (hasReturnValue)
                {
                    Hashtable processResult = Hashtable.Synchronized(new Hashtable());
                    context.ProcessResult = processResult;
                    context.HasReturnValueMethod = hasReturnValueMethod;
                }
                else
                {
                    context.VoidMethod = processMethod;
                }
                context.HasReturnValue = hasReturnValue;
                return context;
            }

            #region 变量/属性

            internal int ThreadCount;

            internal DataWithStateList DataWithStates;

            internal bool NeedWaitAll;

            internal bool HasReturnValue;

            internal WaitCallback VoidMethod;

            internal Func<object, object> HasReturnValueMethod;

            private Hashtable _processResult;
            /// <summary>
            /// 
            /// </summary>
            internal Hashtable ProcessResult
            {
                get { return _processResult; }
                set { _processResult = value; }
            }

            internal void SetReturnValue(object obj, object result)
            {
                lock (_processResult.SyncRoot)
                {
                    _processResult[obj] = result;
                }
            }
            private bool _breakSignal;
            /// <summary>
            /// 
            /// </summary>
            internal void SetBreakSignal()
            {
                if (NeedWaitAll) throw new NotSupportedException("设定为NeedWaitAll时不可设置BreakSignal");
                _breakSignal = true;
            }
            /// <summary>
            /// 
            /// </summary>
            internal bool NeedBreak
            {
                get
                {
                    return !NeedWaitAll && _breakSignal;
                }
            }
            #endregion

            internal void Exec(object obj)
            {
                if (HasReturnValue)
                {
                    SetReturnValue(obj, HasReturnValueMethod(obj));
                }
                else
                {
                    VoidMethod(obj);
                }
                DataWithStates.SetState(obj, ProcessState.Processed);
            }
        }
        /// <summary>
        /// 
        /// </summary>
        private enum ProcessState : byte
        {
            WaitForProcess = 0,
            Processing = 1,
            Processed = 2
        }
        /// <summary>
        /// 
        /// </summary>
        private class DataWithStateList : List<DataWithState>
        {
            public void SetState(object obj, ProcessState state)
            {
                lock (((ICollection)this).SyncRoot)
                {
                    DataWithState dws = this.Find(delegate(DataWithState i) { return Object.Equals(i.Data, obj); });

                    if (dws != null)
                    {
                        dws.State = state;
                    }
                }
            }

            public ProcessState GetState(object obj)
            {
                lock (((ICollection)this).SyncRoot)
                {
                    DataWithState dws = this.Find(delegate(DataWithState i) { return Object.Equals(i.Data, obj); });
                    return dws.State;
                }
            }

            private int GetCount(ProcessState state)
            {
                List<DataWithState> datas = this.FindAll(delegate(DataWithState i) { return i.State == state; });
                if (datas == null) return 0;
                return datas.Count;
            }

            public int WaitForDataCount
            {
                get
                {
                    return GetCount(ProcessState.WaitForProcess);
                }
            }

            internal object GetWaitForObject()
            {
                lock (((ICollection)this).SyncRoot)
                {
                    DataWithState dws = this.Find(delegate(DataWithState i) { return i.State == ProcessState.WaitForProcess; });
                    if (dws == null) return null;
                    dws.State = ProcessState.Processing;
                    return dws.Data;
                }
            }

            internal bool IsWaitForData(object obj, bool setState)
            {
                lock (((ICollection)this).SyncRoot)
                {
                    DataWithState dws = this.Find(delegate(DataWithState i) { return i.State == ProcessState.WaitForProcess; });

                    if (setState && dws != null) dws.State = ProcessState.Processing;

                    return dws != null;
                }
            }
        }
        /// <summary>
        /// 
        /// </summary>
        private class DataWithState
        {
            public readonly object Data;
            public ProcessState State;

            public DataWithState(object data, ProcessState state)
            {
                Data = data;
                State = state;
            }
        }
        /// <summary>
        /// 线程号
        /// </summary>
        private static int _threadNo = 0;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="state"></param>
        private static void DoPrivate(object state)
        {
            object[] objs = state as object[];

            IList datas = objs[0] as IList;
            AsyncContext context = objs[1] as AsyncContext;
            AutoResetEvent evt = objs[2] as AutoResetEvent;

            DataWithStateList objStates = context.DataWithStates;

#if DEBUG 
            Thread.CurrentThread.Name = "Thread " + _threadNo; 
  
            Interlocked.Increment(ref _threadNo); 
            string threadName = Thread.CurrentThread.Name + "[" + Thread.CurrentThread.ManagedThreadId + "]"; 
            Trace.WriteLine("线程ID:" + threadName); 
#endif
            if (datas != null)
            {
                for (int i = 0; i < datas.Count; i++)
                {
                    if (context.NeedBreak)
                    {
#if DEBUG 
                        Trace.WriteLine("线程" + threadName + "未执行完跳出"); 
#endif
                        break;
                    }
                    object obj = datas[i];
                    if (objStates.IsWaitForData(obj, true))
                    {
                        if (context.NeedBreak)
                        {
#if DEBUG 
                            Trace.WriteLine("线程" + threadName + "未执行完跳出"); 
#endif
                            break;
                        }

                        context.Exec(obj);

#if DEBUG 
                        Trace.WriteLine(string.Format("线程{0}处理{1}", threadName, obj)); 
#endif
                    }
                }
            }

            if (context.NeedWaitAll)
            {
                //如果执行完当前进程的数据，还要查看剩下多少没有做，如果还剩下超过ThreadCount个没有做 
                while (objStates.WaitForDataCount > context.ThreadCount)
                {
                    if (context.NeedBreak) break;

                    object obj = objStates.GetWaitForObject();
                    if (obj != null && objStates.IsWaitForData(obj, false))
                    {
                        if (context.NeedBreak)
                        {
#if DEBUG 
                            Trace.WriteLine("线程" + threadName + "未执行完跳出"); 
#endif
                            break;
                        }

                        context.Exec(obj);

#if DEBUG 
                        Trace.WriteLine(string.Format("线程{0}执行另一个进程的数据{1}", threadName, obj)); 
#endif
                    }
                }
            }

            evt.Set();
        }
        #endregion
    }
}
